package com.rp.mqtt;//package com.rp.draw.mqtt;
//
//import cn.hutool.core.util.RandomUtil;
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONException;
//import com.rp.draw.data.data_handle.PumpHandle;
//import com.rp.draw.data.data_handle.ValueHandle;
//import com.rp.draw.data.data_send.server_ws.MyChannelHandlerPool;
//import com.rp.draw.entity.Alarm;
//import com.rp.draw.entity.OperatingRecord;
//import com.rp.draw.service.AlarmService;
//import com.rp.draw.service.CollectorFcService;
//import com.rp.draw.service.OperatingRecordService;
//import com.rp.common.utils.ConvertT;
//import io.netty.channel.Channel;
//import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
//import lombok.Data;
//import lombok.extern.slf4j.Slf4j;
//import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.integration.annotation.IntegrationComponentScan;
//import org.springframework.integration.annotation.ServiceActivator;
//import org.springframework.integration.channel.DirectChannel;
//import org.springframework.integration.core.MessageProducer;
//import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
//import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
//import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
//import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
//import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
//import org.springframework.messaging.Message;
//import org.springframework.messaging.MessageChannel;
//import org.springframework.messaging.MessageHandler;
//import org.springframework.messaging.MessagingException;
//
//import java.util.ArrayList;
//import java.util.LinkedHashMap;
//import java.util.List;
//import java.util.Map;
//
///**
// * 类说明:
// *
// * @author ****
// * @date 2021-03-20 14:08  新建
// */
//@Data
//@Configuration
//@IntegrationComponentScan
//@Slf4j
//public class MqttReceiveConfig1 {
//
//    @Autowired
//    private CollectorFcService collectorFcService;
//    @Autowired
//    private ValueHandle valueHandle;
//
//
//    @Autowired
//    private OperatingRecordService operatingRecordService;
//    @Autowired
//    private AlarmService alarmService;
//
//    /**
//     * 订阅的bean名称
//     */
//    public static final String CHANNEL_NAME_IN = "mqttInboundChannel";
//    /**
//     * 发布的bean名称
//     */
//    public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";
//
//    @Value("${mqtt.enable}")
//    private Boolean enable;
//
//    @Value("${mqtt.username}")
//    private String username;
//
//    @Value("${mqtt.password}")
//    private String password;
//
//    @Value("${mqtt.host}")
//    private String hostUrl;
//
//    @Value("${mqtt.clientId}")
//    private String clientId;
//
//    @Value("${mqtt.topic}")
//    private String defaultTopic;
//
//    @Value("${mqtt.timeout}")
//    private int completionTimeout;
//
//    @Value("${mqtt.keepalive}")
//    private int keepAlive;
//
//    @Value("${spring.profiles.active}")
//    private String active;
//
//    @Bean
//    public MqttConnectOptions getMqttConnectOptions() {
//        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
//        mqttConnectOptions.setCleanSession(true);
//        mqttConnectOptions.setConnectionTimeout(completionTimeout);
//        mqttConnectOptions.setKeepAliveInterval(keepAlive);
//        mqttConnectOptions.setAutomaticReconnect(true);
//        mqttConnectOptions.setUserName(username);
//        mqttConnectOptions.setPassword(password.toCharArray());
//        mqttConnectOptions.setServerURIs(new String[]{hostUrl});
//        return mqttConnectOptions;
//    }
//
//    @Bean
//    public MqttPahoClientFactory mqttClientFactory() {
//        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
//        factory.setConnectionOptions(getMqttConnectOptions());
//        return factory;
//    }
//
//
//
//    /**
//     * MQTT信息通道（生产者）
//     *
//     * @return {@link MessageChannel}
//     */
//    @Bean(name = CHANNEL_NAME_OUT)
//    public MessageChannel mqttOutboundChannel() {
//        return new DirectChannel();
//    }
//
//    /**
//     * MQTT消息处理器（生产者）
//     *
//     * @return {@link MessageHandler}
//     */
//    @Bean
//    @ServiceActivator(inputChannel = CHANNEL_NAME_OUT)
//    public MessageHandler mqttOutbound() {
//        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
//                clientId,
//                mqttClientFactory());
//        messageHandler.setAsync(true);
//        messageHandler.setDefaultTopic(defaultTopic);
//        return messageHandler;
//    }
//
//
//
//    //接收通道
//    @Bean
//    public MessageChannel mqttInputChannel() {
//        return new DirectChannel();
//    }
//
//    //配置client,监听的topic
//    @Bean
//    public MessageProducer inbound() {
//        if(enable){
//            MqttPahoMessageDrivenChannelAdapter adapter =
//                    new MqttPahoMessageDrivenChannelAdapter(clientId + "_" + active + "_" + RandomUtil.randomString(16), mqttClientFactory(),
//                            defaultTopic.split(","));
//            adapter.setCompletionTimeout(completionTimeout);
//            adapter.setConverter(new DefaultPahoMessageConverter());
//            adapter.setQos(1);
//            adapter.setOutputChannel(mqttInputChannel());
//            return adapter;
//        }
//        return null;
//    }
//
//    public static Map<String,Boolean> IdMap = new LinkedHashMap<>();
//
//    //通过通道获取数据
//    @Bean
//    @ServiceActivator(inputChannel = "mqttInputChannel")
//    public MessageHandler handler() {
//        return new MessageHandler() {
//            @Override
//            public void handleMessage(Message<?> message) throws MessagingException {
//                //log.info("主题：{}，消息接收到的数据：{}", message.getHeaders().get("mqtt_receivedTopic"), message.getPayload());
//                String msg = (String) message.getPayload();
//                String topic = (String) message.getHeaders().get("mqtt_receivedTopic");
//                System.out.println(topic);
//                try {
//                    if ("draw".equals(topic)) {
//                        Map parMap = JSON.parseObject(msg, Map.class);
//                        String rw = String.valueOf(parMap.get("rw"));
//                        String data = String.valueOf(parMap.get("data"));
//                        List<Map> list = JSON.parseArray(data, Map.class);
//                        for(Map map : list){
//                            if(rw.equals("read")){
//                                String id = String.valueOf(map.get("id"));
//                                String value = JSON.toJSONString(map.get("value"));
//                                boolean boo = collectorFcService.updateValue(id,value);
//                            }else if(rw.equals("write")){
//                                String id = String.valueOf(map.get("id"));
//                                Integer code = (Integer) map.get("code");
//                                String messageCode = String.valueOf(map.get("message"));
//                                if(code == 200){
//                                    messageCode = "success";
//                                    IdMap.put(id,true);
//                                }
//                                operatingRecordService.updateById(new OperatingRecord(id,code,messageCode));
//                            }
//                        }
//                    }
//                    if("Easystem".equals(topic.split("/")[0])&&"data".equals(topic.split("/")[topic.split("/").length-1])){
//                        System.out.println(msg);
//                        String imei=topic.split("/")[1];
//                        Map parMap = JSON.parseObject(msg,Map.class);
////                        for (int i = 1; i < 11; i++) {
////                            String ad="P0"+i;
////                            if (i==10) ad="P10";
////                            String id = collectorFcService.getIdByImei(imei, Integer.toString(i));
////                            collectorFcService.updateValue(collectorFcService.getIdByImei(imei,Integer.toString(i)),parMap.get(ad).toString());
////                        }
//                        List<Map> p = collectorFcService.getIdByImei(imei, "P");
//
//                        for (Map map:p
//                             ) {
////                            if ()
//                            Alarm alarm = new Alarm();
//                            alarm.setDeviceID(map.get("deviceId").toString());
//                            alarm.setInfoCode("1");
////                            alarm.setStatus();
////                            alarm.setDescription("数值波动过大");
//                            alarm.setNowValue(parMap.get(map.get("address")).toString());
//                            alarm.setLevel("1");
//                            alarm.setStatus("0");
////                            System.out.println("以前值"+map.get("value").toString());
//                            alarm.setLastValue1(map.get("value").toString());
//                            collectorFcService.updateValue(map.get("id").toString(),parMap.get(map.get("address")).toString());
//                            Channel gateway_id = MyChannelHandlerPool.UserChannel.get(map.get("gateway_id").toString());
//                            if (gateway_id != null) {
//                                gateway_id.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(valueHandle.circuit(map.get("id").toString()))));
//                            }
//
//                            if (MyChannelHandlerPool.UserChannel.size()!= 0) {
//                                Channel channel = MyChannelHandlerPool.UserChannel.get(map.get("uid").toString());
//                                System.out.println("数据准备发送了");
//                                if (channel != null) {
//                                    System.out.println("开始判断要不要发送");
////                                    if (Math.abs(Integer.valueOf(alarm.getNowValue())-Integer.valueOf(map.get("value").toString()))>Integer.valueOf(map.get("threshold").toString().split(",")[2])) {
////                                        System.out.println("数据发送了");
//                                        alarmService.ifSave(alarm,channel,map.get("uid").toString());
////                                        channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(alarmService.getOneList(map.get("uid").toString()))));
////                                    }
//                               }
//                            }
//
//
//                        }
//                        List<Map> D = collectorFcService.getIdByImei(imei, "D");
//                        for (int i = 0; i < D.size(); i++) {
//                            ArrayList<String> strings = new ArrayList<>();
//                            StringBuilder builder=new StringBuilder();
//                            String i1=parMap.get("DI"+ ConvertT.intToString(4*i+1)).toString();
//                            System.out.println("i1"+i1);
//                            String i2=parMap.get("DI"+ConvertT.intToString(4*i+2)).toString();
//                            System.out.println("i2"+i2);
//                            String i3=parMap.get("DI"+ConvertT.intToString(4*i+3)).toString();
//                            System.out.println("i3"+i3);
//                            String i4=parMap.get("DI"+ConvertT.intToString(4*i+4)).toString();
//                            System.out.println("i4"+i4);
//                            strings.add(ConvertT.Negate(i1));
//                            strings.add(ConvertT.Negate(i2));
//                            strings.add(i3);
//                            strings.add(i4);
//                            Alarm alarm = new Alarm();
//                            alarm.setDeviceID(D.get(i).get("deviceId").toString());
//                            alarm.setInfoCode("0");
//
////                            alarm.setDescription(builder.toString());
//                            alarm.setNowValue(strings.toString());
//                            String value = D.get(i).get("value").toString();
//                            int flag=0;
//                            if (value == null) {
//                                value=strings.toString();
//                                flag=1;
//                            }
//                            alarm.setLastValue1(value);
//
//                            collectorFcService.updateValue(D.get(i).get("id").toString(),strings.toString());
//                            PumpHandle.ifSend(D.get(i).get("id").toString(),alarm,flag);
//                            if (alarm.getLastValue1().equals(strings.toString()))
//                            if (i1.equals("0")) {
//                                builder.append("断电");
//                                alarm.setStatus("0");
//                                alarm.setLevel("1");
//                                alarm.setDescription("断电异常报警");
//                                alarmService.save(alarm);
//                                Channel uid = MyChannelHandlerPool.UserChannel.get(D.get(i).get("uid").toString());
//                                if (uid != null) {
//                                    uid.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(alarmService.getOneList(D.get(i).get("uid").toString()))));
//                                }Channel gateway_id = MyChannelHandlerPool.UserChannel.get(D.get(i).get("gateway_id").toString());
//                                if (gateway_id != null) {
//                                    gateway_id.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(collectorFcService.getCircuitValue(D.get(i).get("gateway_id").toString()))));
//                                }
//                                }
//
//                            if (i2.equals("0")) {
//                                alarm.setDescription("手动异常报警");
//                                alarm.setStatus("2");
//                                alarmService.save(alarm);
//                                alarm.setLevel("1");
//                                Channel uid = MyChannelHandlerPool.UserChannel.get(D.get(i).get("uid").toString());
//                                if (uid != null) {
//                                    uid.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(alarmService.getOneList(D.get(i).get("uid").toString()))));
//                                }
//
//                                Channel gateway_id = MyChannelHandlerPool.UserChannel.get(D.get(i).get("gateway_id").toString());
//                                if (gateway_id != null) {
//                                    gateway_id.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(collectorFcService.getCircuitValue(D.get(i).get("gateway_id").toString()))));
//                                } }
//                            if (i3.equals("1")) {
//                                alarm.setDescription("故障报警");
//                                alarmService.save(alarm);
//                                alarm.setStatus("3");
//                                Channel uid = MyChannelHandlerPool.UserChannel.get(D.get(i).get("uid").toString());
//                                if (uid != null) {
//                                    uid.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(alarmService.getOneList(D.get(i).get("uid").toString()))));
//                                } Channel gateway_id = MyChannelHandlerPool.UserChannel.get(D.get(i).get("gateway_id").toString());
//                                if (gateway_id != null) {
//                                    gateway_id.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(collectorFcService.getCircuitValue(D.get(i).get("gateway_id").toString()))));
//                                } }
//                            if (i4.equals("0")) {
//                                alarm.setDescription("运行报警");
//                                alarm.setStatus("4");
//                                alarm.setLevel("3");
//                                alarm.setInfoCode("1");
//                                alarmService.save(alarm);
//                                Channel uid = MyChannelHandlerPool.UserChannel.get(D.get(i).get("uid").toString());
//                                if (uid != null) {
//                                    uid.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(alarmService.getOneList(D.get(i).get("uid").toString()))));
//                                }  Channel gateway_id = MyChannelHandlerPool.UserChannel.get(D.get(i).get("gateway_id").toString());
//                                if (gateway_id != null) {
//                                    gateway_id.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(collectorFcService.getCircuitValue(D.get(i).get("gateway_id").toString()))));
//                                } }
////                            MyChannelHandlerPool.UserChannel.get(D.get(i).get("uid").toString()).writeAndFlush("alarmService.getOneList(D.get(i).get(\"uid\").toString())");
//                            System.out.println("mqtt方法执行"+D.get(i).get("deviceId").toString());
//
//
//                        }
//
//                    }
//                } catch (JSONException e) {
//                    log.error("JSON Format Parsing Exception : {}", e);
//                }
//            }
//        };
//    }
//}
//
